package org.unigram.oluolu.recommend;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

/**
 * Append queries to related user pairs. 
 */
public class AddFeaturesReducer extends
        Reducer<LongWritable, Text, LongWritable, Text> {
    
    /**
     * reduce method.
     * 
     * @param related related user with target users
     * @param targetIterator target users
     * @param context write key value into hdfs
     * 
     * @throws IOException -
     * @throws InterruptedException -
     */
    @Override
    public void reduce(final LongWritable related,
            final Iterable<Text> targetIterator,
            final Context context)
            throws IOException, InterruptedException {

        Text rtValue = null;
        List<Long> targets = new LinkedList<Long>();
        for (Text v : targetIterator) {
            /* features in related example? */
            if (v.find(":") >= 0) { 
                rtValue = new Text(related+"\t"+v);
                continue;
            }
            targets.add(Long.parseLong(v.toString()));
        }

        /* output recommendations with target features */
        for (Long target : targets) {
            context.write(new LongWritable(target), rtValue);
        }
    }    
}
